#include "Wakeup.h"
#include "TcpSocketOps.h"
#include "EventLoop.h"
#include "Connector.h"
#include "Channel.h"
#include "Acceptor.h"
#include "Options.h"
#include "Log.h"

//TODO: atomic me
#include "DSTHMutex.h"
namespace {
    dsth::Mutex mutex;
    unsigned short GetNextListenPort() {
        static unsigned short listener_port_ = dsth::GetOption("SelfConnectionBeginPort").usVal;
        dsth::MutexLock lock(&mutex);
        return listener_port_++;
    }
}

namespace dsth {

Wakeup::Wakeup(EventLoop* loop) 
    : loop_(loop),
    self_port_(0),
    is_self_conn_(false),
    wakefd_(0),
    readfd_(0) {
}

Wakeup::~Wakeup() {
    //TODO: TestMe
    if (wakefd_)
        closesocket(wakefd_);
    if (readfd_)
        closesocket(readfd_);
}

void Wakeup::AttachToLoop() {
    if (GetOption("SelfConnectionEable").boolVal) {
        CreateWakeupListener();
    }
}

void Wakeup::CreateWakeupListener() {
    unsigned short port = GetNextListenPort();
    InetSocketAddress addr(GetOption("SelfConnectionHost").strVal, port);
    wakeup_listener_.reset(new Acceptor(loop_, addr));
    wakeup_listener_->Listen();
    wakeup_listener_->SetAcceptHandler(std::make_pair(this, &Wakeup::SelfConnected));
    self_port_ = port;
    ConnectToSelf();
}

void Wakeup::SelfConnected(int sockfd, const InetSocketAddress& peer_addr) {
    readfd_ = sockfd;
    listen_channel_.reset(new Channel(sockfd, loop_));
    listen_channel_->SetReadHandler(std::make_pair(this, &Wakeup::OnReadable));
    listen_channel_->EnableReading();    
}

void Wakeup::ConnectToSelf() {
    InetSocketAddress addr(GetOption("SelfConnectionHost").strVal, self_port_);
    conn_.reset(new Connector(loop_, addr));
    conn_->SetConnectHandler(std::make_pair(this, &Wakeup::HandleSelfConnection));
    conn_->Connect();
}

void Wakeup::HandleSelfConnection(int fd) {
    assert(!is_self_conn_);
    wakefd_ = fd;
    LOG_INFO("[Wakeup::HandleSelfConnection] Self Connection Successful.");
    wake_channel_.reset(new Channel(wakefd_, loop_));
    is_self_conn_ = true;
}

void Wakeup::Waken() {
    if (is_self_conn_) {
        LOG_DEBUG("[Wakeup::Waken]");
        uint64_t one = 1;
        int n = net::Write(wakefd_, (const char*)(&one), sizeof one);
        if (n != sizeof one) {
            LOG_ERROR("[Wakeup::Waken] Write %d", n);
        }
    }
}

void Wakeup::OnReadable(Timestamp time) {
    uint64_t one = 1;
    LOG_DEBUG("[Wakeup::OnReadable]");
    int n = net::Read(readfd_, (char*)&one, sizeof one);
    if (n != sizeof one) {
        LOG_ERROR("[Wakeup::OnReadable] Read %d", n);
    }
}

} // namespace
